/*
 * This file is part of b.AMP.
 *
 * b.AMP is free software: you can redistribute it and/or modify it
 * under the terms of the GNU Lesser General Public License as
 * published by the Free Software Foundation, either version 3 of the
 * License, or (at your option) any later version.
 *
 * b.AMP is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with b.AMP.  If not, see <http://www.gnu.org/licenses/>.
 */

#include "message_ack.hpp"

#include <tr1/functional>
#include <boost/thread.hpp>


bamp::MessageAck::MessageAck(const int count)
    : m_handlerSem   (new Semaphore(count)),
      m_waitSem      (new Semaphore(0)),
      m_refSem       (new Semaphore(0)),
      m_function     (new MessageAckFunc(NULL)),
      m_functionMutex()
{
    // A count of zero means that we are not waiting for anything.
    ack();
}


bamp::MessageAck::~MessageAck()
{
    // Decrease reference count and cleanup when it reaches 0.
    if (!m_refSem->try_wait()) {
        delete m_refSem;
        delete m_waitSem;
        delete m_handlerSem;
        delete m_function;
    }    
}


bamp::MessageAck::MessageAck(const MessageAck& e)
    : m_handlerSem(e.m_handlerSem),
      m_waitSem   (e.m_waitSem),
      m_refSem    (e.m_refSem),
      m_function  (e.m_function)
{
    // Increase reference count
    m_refSem->post();
}


const bamp::MessageAck& bamp::MessageAck::operator=(const MessageAck& e)
{
    if (this != &e) {

        // Decrease reference count and cleanup when it reaches 0.
        if (!m_refSem->try_wait()) {
            delete m_refSem;
            delete m_waitSem;
            delete m_handlerSem;
            delete m_function;
        }

        // Assign semaphores
        m_handlerSem = e.m_handlerSem;
        m_waitSem    = e.m_waitSem;
        m_refSem     = e.m_refSem;
        m_function   = e.m_function;

        // Increase reference count
        m_refSem->post();
    }
    return *this;
}


void bamp::MessageAck::wait()
{
    // Wait. Then, post again to make multiple calls possible.
    m_waitSem->wait();
    m_waitSem->post();
}


void bamp::MessageAck::callback(const MessageAckFunc& function)
{
    // If it is already time to call the function, then call it. Else,
    // save a copy for later.
    if (m_waitSem->try_wait()) {
        function();
        m_waitSem->post();
    } else {
        boost::mutex::scoped_lock lock(m_functionMutex);
        (*m_function) = function;
    }
}


void bamp::MessageAck::ack()
{
    // Decrease handler count. When it reaches zero, make wait()
    // non-blocking. Call the configured callback, if any.
    if (!m_handlerSem->try_wait()) {
        boost::mutex::scoped_lock lock(m_functionMutex);
        if (*m_function) {
            MessageAckFunc funcCopy(*m_function);
            lock.unlock();
            funcCopy();
        }
        m_waitSem->post();
    }
}
